package com.atguigu.pro3

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.{TimeCharacteristic, TimerService}
import org.apache.flink.util.Collector

/**
 * @description: individual patterns,个体模式可以包括单例模式和循环模式,单例模式只接收一个事件,而循环模式可以接收多个,可以在一个个体模式后追加两次,也就是指定循环次数
 * @time: 2021/4/6 14:30 订单超时事件
 * @author: baojinlong
 * */


object CepDemo02WithState {
  def main(args: Array[String]): Unit = {
    /**
     * 匹配出现4次 start.times(4)     匹配出现2,3或者4次,并且尽可能多地重复 start.times(2,4).greedy
     * 匹配出现0或者4次 start.times(4).optional 匹配出现1次或多次start.oneOrMore
     * 匹配出现2,3或者4次 start.times(2,4) 匹配出现0次,2次或多次,并且尽可能多地重复 start.timesOrMore(2).optional.greedy
     * 个体模式的条件,每隔模式都需要指定触发条件,作为模式是否接受事件进入的判断依据
     * cep中的个体模式主要通过调用.where or until来指定条件
     * 严格近邻 strict contiguity 宽松relaxed 非确定性宽松紧邻non-deterministic relaxed contiguity
     * 匹配事件的提取
     */
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 不影响输出顺序
    environment.setParallelism(4)
    // 有事件时间的都设置这个时间
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 0.从文件中读取数据
    val orderEventStream: DataStream[OrderEvent] = environment.readTextFile("src/main/resources/OrderLog.csv")
      .map(data => {
        val arr: Array[String] = data.split(",")
        OrderEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000)

    // 自定义ProcessFunction进行复杂事件检测
    val orderResultStream: DataStream[OrderResult] = orderEventStream
      .keyBy(_.orderId)
      .process(new OrderPayMatchResult)

    orderResultStream.print("success")
    orderResultStream.getSideOutput(new OutputTag[OrderResult]("timeout")).print("timeout")
    environment.execute("order timeout job")
  }
}

class OrderPayMatchResult extends KeyedProcessFunction[Long, OrderEvent, OrderResult] {
  // 定义状态,考虑到乱序数据 pay和create的数据处理的顺序是不确定的,需要标识位标识create,pay是否已经来过,定时器时间戳
  lazy val isCreatedState: ValueState[Boolean] = getRuntimeContext.getState[Boolean](new ValueStateDescriptor[Boolean]("isCreatedState", classOf[Boolean]))
  lazy val isPayedState: ValueState[Boolean] = getRuntimeContext.getState[Boolean](new ValueStateDescriptor[Boolean]("isPayedState", classOf[Boolean]))
  lazy val timerTsState: ValueState[Long] = getRuntimeContext.getState[Long](new ValueStateDescriptor[Long]("timerTsState", classOf[Long]))
  // 定义侧输出流标签
  val orderTimeoutOutputTag = new OutputTag[OrderResult]("timeout")

  override def processElement(value: OrderEvent, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#Context, out: Collector[OrderResult]): Unit = {
    // 获取timerService
    val timerService: TimerService = ctx.timerService
    // 先拿到当前状态
    val isCreatedStateValue: Boolean = isPayedState.value
    val isPayedStateValue: Boolean = isPayedState.value
    val timerTsStateValue: Long = timerTsState.value
    // 来的是create,要继续判断是否pay过
    val nowDataTs: Long = value.timestamp * 1000
    val eventType: String = value.eventType
    if ("create".equals(eventType)) {
      // 如果已经支付过,正常支付,输出匹配成功的结果
      if (isPayedStateValue) {
        out.collect(OrderResult(value.orderId, "payed successfully"))
        // 已经处理完毕清空状态
        isCreatedState.clear()
        isPayedState.clear()
        timerTsState.clear()
        timerService.deleteEventTimeTimer(timerTsStateValue)
      } else {
        // 如果还没pay过.注册定时器等待15分钟
        val ts: Long = nowDataTs + 900 * 1000
        timerService.registerEventTimeTimer(ts)
        // 更新状态
        timerTsState.update(ts)
        isCreatedState.update(true)
      }
    } else if ("pay".equals(eventType)) {
      // 如果当前来的是pay,要判断是否create过
      if (isCreatedStateValue) {
        // 如果已经create过,匹配成功,还要判断一下pay时间是否已经超过了定时器时间
        if (nowDataTs < timerTsStateValue) {
          // 没有超时则正常输出
          out.collect(OrderResult(value.orderId, "payed successfully"))
        } else {
          // 已经超时,输出超时
          ctx.output(orderTimeoutOutputTag, OrderResult(value.orderId, "payed but already timeout"))
        }
        // 只要输出结果,当前order处理已经结束,清空状态和定时器
        isCreatedState.clear()
        isPayedState.clear()
        timerTsState.clear()
        timerService.deleteEventTimeTimer(timerTsStateValue)
      } else {
        // 如果create没来,注册定时器,等到pay的时间就可以
        timerService.registerEventTimeTimer(nowDataTs)
        // 更新状态
        timerTsState.update(nowDataTs)
        isPayedState.update(true)
      }
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#OnTimerContext, out: Collector[OrderResult]): Unit = {
    // 获取当前订单id
    val orderId: Long = ctx.getCurrentKey
    // 定时器触发,pay来了,没等到create
    if (isPayedState.value) {
      // 如果这个值仍然是true则肯定是pay没有等到create
      ctx.output(orderTimeoutOutputTag, OrderResult(orderId, "payed but not found create log"))
    } else {
      // create来了,没有pay订单支付超时
      ctx.output(orderTimeoutOutputTag, OrderResult(orderId, "order timeout"))
    }
    // 清空所有状态,定时器触发之后就不要再删除定时器
    isCreatedState.clear()
    isPayedState.clear()
    timerTsState.clear()
  }
}
